package com.learn.spark.analysis

import com.learn.spark.util.DateUtil
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object SalesCountGroupByDate {
  def main(args: Array[String]): Unit = {
    if (args.length != 1) {
      throw new RuntimeException("must give action name(action name must be collect/cart/click/alipay)");
    }
    val ACTION = args(0)
    val sourceFile = "hdfs://master:8020/hadoop/result/tmall/product_" + ACTION + "_top_10/*.csv"

    val files = "hdfs://master:8020/hadoop/log/tianchi_2014002_rec_tmall_log_part*.txt"
    val SEPERATE = "\\u0001";
    val sparkSession: SparkSession = SparkSession.builder().appName("top_1_sales_count").getOrCreate()
    val context: SparkContext = sparkSession.sparkContext

    val item_id: String = context.textFile(sourceFile).filter(x => (!x.contains("count"))).map(x => {
      val arr = x.split(",")
      (arr(1).toInt, arr(0))
    })
      .sortByKey(false)
      .take(1).head._2
    val APP_NAME = "top_1_sales_count_" + item_id
    val topNResult: Array[(String, Int)] = context.textFile(files)
      .filter(s => s.contains(item_id))
      .map(x => (DateUtil.parseDate(x.split(SEPERATE)(3)), 1))
      .reduceByKey(_ + _)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .map(y => (y._2, y._1))
    val resultRdd: RDD[(String, Int)] = context.parallelize(topNResult).repartition(1)
    val rowRdd: RDD[Row] = resultRdd.map(tup => Row(item_id, tup._1, tup._2))
    val schema = StructType(Array(
      StructField("item_id", StringType, true),
      StructField("month", StringType, true),
      StructField("count", IntegerType, true)
    ))
    val stuDf: DataFrame = sparkSession.createDataFrame(rowRdd, schema)
    stuDf.repartition(1)
      .write
      .mode("overwrite")
      .option("header", "true")
      .option("delimiter", ",")
      .csv("hdfs://master:8020/hadoop/result/tmall/" + APP_NAME)
    context.stop()
  }
}
